#include "work_queue.h"
#include <assert.h>

typedef struct _THREAD_CONTEXT {
  WorkQueue *work_queue_;
  void *thread_data_;
} THREAD_CONTEXT, *PTHREAD_CONTEXT;

bool WorkQueue::create(const unsigned int num_of_threads,
                       void **thread_data /*=NULL*/) {
  work_item_queue_ = new WorkItemQueue();
  if (NULL == work_item_queue_) {
    return false;
  }
  sync_obj_array_[SEMAPHORE_INDEX] = CreateSemaphore(NULL, 0, LONG_MAX, NULL);
  if (sync_obj_array_[SEMAPHORE_INDEX] == NULL) {
    delete work_item_queue_;
    work_item_queue_ = NULL;
    return false;
  }
  sync_obj_array_[ABORT_EVENT_INDEX] = CreateEvent(NULL, TRUE, FALSE, NULL);
  if (sync_obj_array_[ABORT_EVENT_INDEX] == NULL) {
    delete work_item_queue_;
    work_item_queue_ = NULL;
    CloseHandle(sync_obj_array_[SEMAPHORE_INDEX]);
    return false;
  }
  InitializeCriticalSection(&cs_);
  threads_ = new HANDLE[num_of_threads];
  if (threads_ == NULL) {
    delete work_item_queue_;
    work_item_queue_ = NULL;
    CloseHandle(sync_obj_array_[SEMAPHORE_INDEX]);
    CloseHandle(sync_obj_array_[ABORT_EVENT_INDEX]);
    DeleteCriticalSection(&cs_);
    return false;
  }
  num_of_threads_ = num_of_threads;
  DWORD thread_id;
  PTHREAD_CONTEXT thread_context;
  for (unsigned int i = 0; i < num_of_threads; i++) {
    thread_context = new THREAD_CONTEXT;
    thread_context->work_queue_ = this;
    thread_context->thread_data_ = thread_data == NULL ? NULL : thread_data[i];
    threads_[i] = CreateThread(NULL, 0, WorkQueue::thread_func, thread_context,
                               0, &thread_id);
    if (threads_[i] == NULL) {
      delete thread_context;
      num_of_threads_ = i;
      destroy(5);
      return false;
    }
  }
  return true;
}

bool WorkQueue::insert_work_item(WorkItemBase *work_iterm) {
  assert(work_iterm != NULL);
  EnterCriticalSection(&cs_);
  work_item_queue_->push(work_iterm);
  LeaveCriticalSection(&cs_);
  if (!ReleaseSemaphore(sync_obj_array_[SEMAPHORE_INDEX], 1, NULL)) {
    assert(false);
    return false;
  }
  return true;
}

WorkItemBase *WorkQueue::remove_work_item() {
  WorkItemBase *work_item;
  EnterCriticalSection(&cs_);
  work_item = work_item_queue_->front();
  work_item_queue_->pop();
  LeaveCriticalSection(&cs_);
  assert(work_item != NULL);
  return work_item;
}

unsigned long __stdcall WorkQueue::thread_func(void *param) {
  PTHREAD_CONTEXT thread_context = (PTHREAD_CONTEXT)param;
  WorkItemBase *work_item = NULL;
  WorkQueue *work_queue = thread_context->work_queue_;
  void *thread_data = thread_context->thread_data_;
  DWORD wait_result;
  for (;;) {
    wait_result = WaitForMultipleObjects(
        NUMBER_OF_SYNC_OBJ, work_queue->sync_obj_array_, FALSE, INFINITE);
    switch (wait_result - WAIT_OBJECT_0) {
    case ABORT_EVENT_INDEX:
      delete thread_context;
      return 0;
    case SEMAPHORE_INDEX:
      work_item = work_queue->remove_work_item();
      if (work_item == NULL) {
        assert(false);
        break;
      }
      work_item->work(thread_data);
      break;
    default:
      assert(false);
      delete thread_context;
      return 0;
    }
  }
  delete thread_context;
  return 1;
}

int WorkQueue::get_threads_num() { return num_of_threads_; }

void WorkQueue::suspend() {
  DWORD code = 0;
  for (unsigned int i = 0; i < num_of_threads_; i++) {
    GetExitCodeThread(threads_[i], &code);
    if (code == STILL_ACTIVE) {
      SuspendThread(threads_[i]);
    }
  }
}

void WorkQueue::resume() {
  DWORD code = 0;
  for (unsigned int i = 0; i < num_of_threads_; i++) {
    GetExitCodeThread(threads_[i], &code);
    if (code == STILL_ACTIVE) {
      ResumeThread(threads_[i]);
    }
  }
}

size_t WorkQueue::get_work_queue_size() {
  EnterCriticalSection(&cs_);
  size_t size = work_item_queue_->size();
  LeaveCriticalSection(&cs_);
  return size;
}

void WorkQueue::destroy(int wait_seconds) {
  if (threads_ == nullptr) {
    return;
  }
  while (0 != get_work_queue_size()) {
    Sleep(wait_seconds * 1000);
  }
  if (!SetEvent(sync_obj_array_[ABORT_EVENT_INDEX])) {
    assert(false);
    return;
  }
  WaitForMultipleObjects(num_of_threads_, threads_, true, INFINITE);
  while (!work_item_queue_->empty()) {
    work_item_queue_->front()->abort();
    work_item_queue_->pop();
  }
  delete work_item_queue_;
  work_item_queue_ = NULL;
  CloseHandle(sync_obj_array_[SEMAPHORE_INDEX]);
  CloseHandle(sync_obj_array_[ABORT_EVENT_INDEX]);
  DeleteCriticalSection(&cs_);
  for (unsigned int i = 0; i < num_of_threads_; i++)
    CloseHandle(threads_[i]);
  delete[] threads_;
}
